package com.atguigu.edu.realtime.app.func;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.realtime.beans.TableProcess;
import jdk.nashorn.internal.codegen.MapCreator;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;

import javax.naming.OperationNotSupportedException;
import java.sql.*;
import java.util.*;

/**
 * @ClassName BroadcastFunction
 * @Description TODO
 * @Author$ 邢家俊
 * @Date 2023-5-6 10:04
 * @Version 1.0
 **/
public class BroadcastFunction extends BroadcastProcessFunction<JSONObject, String, JSONObject> {
    private MapStateDescriptor<String, TableProcess> mapStateDescriptorDS;
    private HashMap<String,TableProcess> hashMapConfig=new HashMap<>();

    public BroadcastFunction(MapStateDescriptor<String, TableProcess> mapStateDescriptorDS) {
        this.mapStateDescriptorDS = mapStateDescriptorDS;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
       // Connection con = DriverManager.getConnection("jdbc:mysql://hadoop101:3306/edu_config?\" +\n" +
       //         "            \"user=root&password=000000&useUnicode=true&\" +\n" +
       //         "            \"characterEncoding=utf8&serverTimeZone=Asia/Shanghai&useSSL=false");
        Connection con = DriverManager.getConnection("jdbc:mysql://hadoop101:3306/edu_config?" +
                "user=root&password=000000&useUnicode=true&" +
                "characterEncoding=utf8&serverTimeZone=Asia/Shanghai&useSSL=false");
        PreparedStatement ps = con.prepareStatement("select * from edu_config.table_process where sink_type = 'dwd'");
        ResultSet rs = ps.executeQuery();
        ResultSetMetaData rsmetaData = rs.getMetaData();
        JSONObject jsonObj = new JSONObject();
        while (rs.next()) {
            for (int i = 1; i <= rsmetaData.getColumnCount(); i++) {
                String columnClassName = rsmetaData.getColumnClassName(i);
                String columnClassvalue = rs.getString(i);
                jsonObj.put(columnClassName, columnClassvalue);
            }
        }

            TableProcess tableProcess = jsonObj.toJavaObject(TableProcess.class);
            String sourceTable = tableProcess.getSourceTable();
            String sinkType = tableProcess.getSinkType();
            String key = sourceTable+":"+sinkType;
            hashMapConfig.put(key,tableProcess);
            //关闭资源
            rs.close();
            ps.close();
            con.close();

    }
    @Override
    public void processElement(JSONObject jsonObj, ReadOnlyContext ctx, Collector<JSONObject> out) throws Exception {
        //{"database":"edu","xid":74996,"data":{"coupon_reduce":0.00,"create_time":"2023-05-06 10:11:16","session_id":
        // "8d73a5a2-f5c8-490f-851e-a548d03c94d4","expire_time":"2023-05-06 10:26:16","order_status":"1002",
        // "origin_amount":600.00,"out_trade_no":"515199566275462","user_id":4,"province_id":27,"trade_body":"Java8新特性等3件商品",
        // "final_amount":600.00,"id":46767},"old":{"order_status":"1001"}
        // ,"commit":true,"type":"update","table":"order_info","ts":1683339076}
        String table = jsonObj.getString("table");
        String type = jsonObj.getString("type");
        String key =table+":"+type;
        ReadOnlyBroadcastState<String, TableProcess> broadcastStateDS = ctx.getBroadcastState(mapStateDescriptorDS);
        TableProcess tableProcess = null;
        JSONObject dateObj = jsonObj.getJSONObject("data");
        if((tableProcess=broadcastStateDS.get(key))!= null || (tableProcess= hashMapConfig.get(key))!= null){
            String sinkColumns = tableProcess.getSinkColumns();
            filterColumn(sinkColumns,dateObj);
            String sinkTable = tableProcess.getSinkTable();
            dateObj.put("sink_table",sinkTable);
            String ts = jsonObj.getString("ts");
            dateObj.put("ts",ts);
            out.collect(dateObj);
        }
    }

    private void filterColumn(String sinkColumns, JSONObject dateObj) {
        String[] sinkColumn = sinkColumns.split(",");
        List<String> stringList = Arrays.asList(sinkColumn);
        Set<Map.Entry<String, Object>> entries = dateObj.entrySet();
        entries.removeIf(entry->!stringList.contains(entry.getKey()));
    }

    @Override
    public void processBroadcastElement(String value, Context ctx, Collector<JSONObject> out) throws Exception {
        // {"before":null,"after":{"source_table":"video_info","source_type":"ALL","sink_table":"dim_video_info","sink_type":"dim","sink_columns":"id,video_name,during_sec,video_status,video_size,video_url,video_source_id,version_id,chapter_id,course_id,publisher_id,create_time,update_time,deleted","sink_pk":"id","sink_extend":null},"source":{"version":"1.5.4.Final","connector":"mysql","name":"mysql_binlog_source","ts_ms":1683339460755,"snapshot":"false","db":"edu_config","sequence":null,"table":"table_process","server_id":0,"gtid":null,"file":"","pos":0,"row":0,"thread":null,"query":null},"op":"r","ts_ms":1683339460756,"transaction":null}
        JSONObject jsonObj = JSON.parseObject(value);
        String sourceTable = jsonObj.getString("source_table");
        BroadcastState<String, TableProcess> broadcastState = ctx.getBroadcastState(mapStateDescriptorDS);
        TableProcess tableProcess = broadcastState.get(sourceTable);
        String op = jsonObj.getString("op");
        if("d".equals(op)){
            JSONObject before = jsonObj.getJSONObject("before");
            TableProcess beforetableProcess = JSON.toJavaObject(before, TableProcess.class);
            String sinkType = beforetableProcess.getSinkType();
            if("dwd".equals(sinkType)){
                String sourTable = beforetableProcess.getSourceTable();
                String sourceType = beforetableProcess.getSourceType();
                String key = sourTable + ":" + sourceType;
                broadcastState.remove(key);
                hashMapConfig.remove(key);
            }
        }else{
            JSONObject after = jsonObj.getJSONObject("after");
            TableProcess  aftertableProcess = JSON.toJavaObject(after, TableProcess.class);
            String sinkType = aftertableProcess.getSinkType();
            if("dwd".equals(sinkType)){
                String sourTable = aftertableProcess.getSourceTable();
                String sourceType = aftertableProcess.getSourceType();
                String key = sourTable + ":" + sourceType;
                broadcastState.put(key,aftertableProcess);
                hashMapConfig.put(key,aftertableProcess);
            }
        }
    }
}
